package heap;


import java.io.*;
import java.util.*;

import Transaction.Transaction;

import exceptions.DBInvalidSlotNumberException;
import exceptions.HFBufMgrException;
import exceptions.HFDiskMgrException;

import Common.GlobalConst;
import Common.GoggleDB;

import storage.data.Page;
import storage.data.PageID;

interface FileType {
	int Temp = 0;
	int ORDINARY = 1;
}

public class Heapfile implements FileType, GlobalConst {
	PageID firstDirPageID;

	int filetype;

	private boolean fileDeleted;

	public String fileName;

	private static int tempfilecount = 0;

	private PageID newPage(int num) throws HFBufMgrException {
		PageID pageId = new PageID();

		try {
			pageId = GoggleDB.bufferMgr.allocatePage(num);
		} catch (Exception e) {
			e.printStackTrace();
			throw new HFBufMgrException(e);
		}
		return pageId;
	}

	private void pinPage(PageID pageno, Page page, boolean emptyPage)
			throws HFBufMgrException {
		try {
			GoggleDB.bufferMgr.pinPage(pageno, page, emptyPage);
		} catch (Exception e) {
			throw new HFBufMgrException(e);
		}

	}

	private void unpinPage(PageID pageno, boolean dirty)
			throws HFBufMgrException {
		try {
			GoggleDB.bufferMgr.unpinPage(pageno, dirty);
		} catch (	Exception e) {
			throw new HFBufMgrException(e);
		}

	}

	private void freePage(PageID pageId) throws HFBufMgrException {
		try {
			GoggleDB.bufferMgr.deallocatePage(pageId,1);
		} catch (Exception e) {
			throw new HFBufMgrException(e);
		}
	}

	/**
	 * get a new datapage from the buffer manager
	 * 
	 * @param dpin
	 * @return
	 * @throws IOException
	 * @throws HFBufMgrException
	 */
	private HFPage newDataPage(DataPageInfo dpin,Transaction tx,String table) throws Exception,
			HFBufMgrException {

		Page apage = new Page();
		PageID pageId = new PageID();
		pageId = newPage(1);

		HFPage hfpage = new HFPage();
		// GoggleDB.bufferMgr.printBuffer();
		pinPage(pageId, apage, true);
		// System.out.println("Pin :"+pageId.pageId);
		// GoggleDB.bufferMgr.printBuffer();
		hfpage.init(pageId, apage,tx,table);

		dpin.pageId.pageId = pageId.pageId;
		dpin.reccnt = 0;
		dpin.availspace = hfpage.availableSpace(tx,table);

		return hfpage;
	}

	private boolean findDataPage(RID rid, PageID dirPageId, HFPage dirpage,
			PageID dataPageId, HFPage datapage, RID rpDataPageRid,Transaction tx,String table)
			throws DBInvalidSlotNumberException, HFBufMgrException,
			IOException, Exception {
		PageID currentDirPageId = new PageID(firstDirPageID.pageId);

		HFPage currentDirPage = new HFPage();
		HFPage currentDataPage = new HFPage();
		RID currentDataPageRid = new RID();
		PageID nextDirPageId = new PageID();

		pinPage(currentDirPageId, currentDirPage, false);

		Tuple atuple = null;

		while (currentDirPageId.pageId != INVALID) {
			for (currentDataPageRid = currentDirPage.firstRecord(tx,table); currentDataPageRid != null; currentDataPageRid = currentDirPage
					.nextRecord(currentDataPageRid,tx,table)) {
				try {
					atuple = currentDirPage.getRecord(currentDataPageRid,tx,table);
				} catch (DBInvalidSlotNumberException e) {
					return false;
				}

				DataPageInfo dpin = new DataPageInfo(atuple,tx,table);
				try {
					pinPage(dpin.pageId, currentDataPage, false);
				} catch (Exception e) {
					unpinPage(currentDirPageId, false/* dirty */);
					dirpage = null;
					datapage = null;
					throw e;
				}

				if (dpin.pageId.pageId == rid.pageNo.pageId) {
					atuple = currentDataPage.returnRecord(rid,tx,table);
					dirpage.setData(currentDirPage.getData());
					dirPageId.pageId = currentDirPageId.pageId;

					datapage.setData(currentDataPage.getData());
					dataPageId.pageId = dpin.pageId.pageId;

					rpDataPageRid.pageNo.pageId = currentDataPageRid.pageNo.pageId;
					rpDataPageRid.slotNo = currentDataPageRid.slotNo;
					return true;
				} else {
					unpinPage(dpin.pageId, false);
				}
			}

			nextDirPageId = currentDirPage.getNextPage(tx,table);
			unpinPage(currentDirPageId, false);
			currentDirPageId.pageId = nextDirPageId.pageId;
			if (currentDirPageId.pageId != INVALID) {
				pinPage(currentDirPageId, currentDirPage, false);
			}
		}

		dirPageId.pageId = dataPageId.pageId = INVALID;
		return false;
	}

	public Heapfile(String name,Transaction tx) throws Exception {
		
		fileDeleted = true;
		fileName = null;
		boolean notFound = false;
		if (name == null) {
			fileName = "tempHeapFile";
			String filenum = Integer.toString(tempfilecount);
			fileName = fileName + filenum;
			filetype = Temp;
			tempfilecount++;
		} else {
			fileName = name;
			filetype = ORDINARY;
		}
		firstDirPageID = null;
		if (filetype == ORDINARY) {
			try {
				firstDirPageID = GoggleDB.storageInterface.get_db_by_name(fileName);
			} catch (Exception e) {
				notFound = true;
			}
		}
		if (notFound) {
			//System.out.println("haha1");
			//System.out.println(fileName);
			firstDirPageID = GoggleDB.storageInterface.add_db(tx,fileName);
			//System.out.println("haha2");
			// check error(new exception: Could not add file entry
			Page apage = new Page();
			pinPage(firstDirPageID, apage, true);
			HFPage firstDirPage = new HFPage();
			firstDirPage.init(firstDirPageID, apage,tx,name);
			PageID pageId = new PageID(INVALID);

			firstDirPage.setNextPage(pageId,tx,name);
			firstDirPage.setPrevPage(pageId,tx,name);
			GoggleDB.bufferMgr.unpinPage(firstDirPageID, true);

		}
		fileDeleted = false;
	}

	public RID insertRecord(byte[] recPtr,Transaction tx,String table) throws Exception,
			HFBufMgrException, DBInvalidSlotNumberException {

		int recLen = recPtr.length;
		boolean found;
		RID curDataPRID = new RID();
		Page pageinbuffer = new Page();
		HFPage curDirPage = new HFPage();
		HFPage curDataPage = new HFPage();

		HFPage nextDirPage = new HFPage();
		PageID curDirPID = new PageID(firstDirPageID.pageId);

		PageID nextDirPageID = new PageID();

		//curDirPage.curPage = curDirPID;
		pinPage(curDirPID, curDirPage, false);

		found = false;
		Tuple atuple;
		DataPageInfo dpinfo = new DataPageInfo();

		// System.out.println("START "+start);
		while (found == false) {
			// System.out.println("START "+start);
			for (curDataPRID = curDirPage.firstRecord(tx,table); curDataPRID != null; curDataPRID = curDirPage
					.nextRecord(curDataPRID,tx,table)) {
				atuple = curDirPage.getRecord(curDataPRID,tx,table);

				dpinfo = new DataPageInfo(atuple,tx,table);
				if (recLen <= dpinfo.availspace) {
					found = true;
					break;
				}
			}
			// System.out.println("found "+found);
			if (found == false) {
				if (curDirPage.availableSpace(tx,table) >= DataPageInfo.size) {
					curDataPage = newDataPage(dpinfo,tx,table);
					// System.out.println("Pin curDirPID"+curDataPage.curPage.pageId);
					atuple = dpinfo.convertToTuple(tx,table,dpinfo.pageId);
					byte[] tmpData = atuple.getTupleByteArray(tx);
					curDataPRID = curDirPage.insertRecord(tmpData,tx,table,curDirPID);
					RID tmprid = curDirPage.firstRecord(tx,table);
					found = true;
				} else {
					nextDirPageID = curDirPage.getNextPage(tx,table);
					if (nextDirPageID.pageId != INVALID) {

						unpinPage(curDirPID, false);

						curDirPID.pageId = nextDirPageID.pageId;

						pinPage(curDirPID, curDirPage, false);
					} else {
						// System.out.println("hfkjshfhsjfhj");
						nextDirPageID = newPage(1);
						pinPage(nextDirPageID, pageinbuffer, true);
						nextDirPage.init(nextDirPageID, pageinbuffer,tx,table);
						PageID tempID = new PageID(INVALID);
						nextDirPage.setNextPage(tempID,tx,table);
						nextDirPage.setCurPage(curDirPID,tx,table);

						curDirPage.setNextPage(nextDirPageID,tx,table);

						unpinPage(curDirPID, true/* true */);

						curDirPID.pageId = nextDirPageID.pageId;
						curDirPage = new HFPage(nextDirPage);
					}
				}
			} else {
				// System.out.println("Pin dpinfo "+dpinfo.pageId.pageId);
				// GoggleDB.bufferMgr.printBuffer();
				//curDataPage.curPage = dpinfo.pageId;
				pinPage(dpinfo.pageId, curDataPage, false);
			}
		}

		RID rid;
		rid = curDataPage.insertRecord(recPtr,tx,table,dpinfo.pageId);

		dpinfo.reccnt++;
		dpinfo.availspace = curDataPage.availableSpace(tx,table);
		// System.out.println("unpin:dpinfo "+dpinfo.pageId.pageId);
		// GoggleDB.bufferMgr.printBuffer();
		unpinPage(dpinfo.pageId, true);
		atuple = curDirPage.returnRecord(curDataPRID,tx,table);
		DataPageInfo dpinfo_ondirpage = new DataPageInfo(atuple,tx,table);
		dpinfo_ondirpage.availspace = dpinfo.availspace;
		dpinfo_ondirpage.reccnt = dpinfo.reccnt;
		dpinfo_ondirpage.pageId.pageId = dpinfo.pageId.pageId;
		dpinfo_ondirpage.flushToTuple(tx,table,curDirPage.curPage);
		//System.out.println(curDirPID.pageId);
		unpinPage(curDirPID, true);

		return rid;
	}

	public boolean deleteRecord(RID rid,Transaction tx,String table) throws IOException, HFBufMgrException,
			DBInvalidSlotNumberException, Exception {
		boolean status;
		HFPage curDirPage = new HFPage();
		PageID curDirPID = new PageID();
		HFPage curDataPage = new HFPage();
		PageID curDataPID = new PageID();
		RID curDataPRID = new RID();

		status = findDataPage(rid, curDirPID, curDirPage, curDataPID,
				curDataPage, curDataPRID,tx,table);
		if (status == false) {
			return status;
		}
		// get datapageinfo from the current directory page:
		Tuple atuple;

		atuple = curDirPage.returnRecord(curDataPRID,tx,table);
		DataPageInfo pdpinfo = new DataPageInfo(atuple,tx,table);

		curDataPage.deleteRecord(rid,tx,table,curDataPage.curPage);
		pdpinfo.reccnt--;
		pdpinfo.flushToTuple(tx,table,pdpinfo.pageId);
		if (pdpinfo.reccnt >= 1) {
			pdpinfo.availspace = curDataPage.availableSpace(tx,table);
			pdpinfo.flushToTuple(tx,table,pdpinfo.pageId);
			unpinPage(curDataPID, true);
			unpinPage(curDirPID, true);
		} else {
			unpinPage(curDataPID, false);
			freePage(curDataPID);
			curDirPage.deleteRecord(curDataPRID,tx,table,curDirPage.curPage);
			curDataPRID = curDirPage.firstRecord(tx,table);

			PageID pageID = curDirPage.getPrevPage(tx,table);
			if ((curDataPRID == null) && (pageID.pageId != INVALID)) {
				HFPage prevDirPage = new HFPage();
				pinPage(pageID, prevDirPage, false);

				pageID = curDirPage.getNextPage(tx,table);
				prevDirPage.setNextPage(pageID,tx,table);
				pageID = curDirPage.getPrevPage(tx,table);
				unpinPage(pageID, true /* = DIRTY */);

				// set prevPage-pointer of next Page
				pageID = curDirPage.getNextPage(tx,table);
				if (pageID.pageId != INVALID) {
					HFPage nextDirPage = new HFPage();
					pageID = curDirPage.getNextPage(tx,table);
					pinPage(pageID, nextDirPage, false);

					// nextDirPage.openHFpage(apage);

					pageID = curDirPage.getPrevPage(tx,table);
					nextDirPage.setPrevPage(pageID,tx,table);
					pageID = curDirPage.getNextPage(tx,table);
					unpinPage(pageID, true /* = DIRTY */);

				}

				// delete empty directory page: (automatically unpinned?)
				unpinPage(curDirPID, false/* undirty */);
				freePage(curDirPID);
			} else {
				unpinPage(curDirPID, true /* == DIRTY */);
			}
		}

		return true;
	}

	public boolean updateRecord(RID rid, Tuple newtuple,Transaction tx,String table)
			throws DBInvalidSlotNumberException, HFBufMgrException, Exception {
		boolean status;
		HFPage dirPage = new HFPage();
		PageID currentDirPageId = new PageID();
		HFPage dataPage = new HFPage();
		PageID currentDataPageId = new PageID();
		RID currentDataPageRid = new RID();

		status = findDataPage(rid, currentDirPageId, dirPage,
				currentDataPageId, dataPage, currentDataPageRid,tx,table);

		if (status != true)
			return status; // record not found
		Tuple atuple = null;
		atuple = dataPage.returnRecord(rid,tx,table);

		// Assume update a record with a record whose length is equal to
		// the original record

		if (newtuple.getLength() != atuple.getLength()) {
			unpinPage(currentDataPageId, false /* undirty */);
			unpinPage(currentDirPageId, false /* undirty */);
		}

		// new copy of this record fits in old space;
		atuple.tupleCopy(newtuple,tx,table,dataPage.curPage);
		unpinPage(currentDataPageId, true /* = DIRTY */);
		unpinPage(currentDirPageId, false /* undirty */);

		return true;
	}

	private void delete_db(String filename,Transaction tx) throws HFDiskMgrException {

		try {
			GoggleDB.storageInterface.del_db(tx,filename);
		} catch (Exception e) {
			throw new HFDiskMgrException(e,
					"Heapfile.java: delete_file_entry() failed");
		}

	}

	public void deleteFile(Transaction tx,String table) throws Exception {
		if (fileDeleted)
			// throw new FileAlreadyDeletedException(null, "file alread
			// deleted");

			// Mark the deleted flag (even if it doesn't get all the way done).
			fileDeleted = true;

		// Deallocate all data pages
		PageID currentDirPageId = new PageID();
		currentDirPageId.pageId = firstDirPageID.pageId;
		PageID nextDirPageId = new PageID();
		nextDirPageId.pageId = 0;

		HFPage currentDirPage = new HFPage();
		Tuple atuple;

		pinPage(currentDirPageId, currentDirPage, false);
		// currentDirPage.openHFpage(pageinbuffer);

		RID rid = new RID();
		while (currentDirPageId.pageId != INVALID) {
			for (rid = currentDirPage.firstRecord(tx,table); rid != null; rid = currentDirPage
					.nextRecord(rid,tx,table)) {
				atuple = currentDirPage.getRecord(rid,tx,table);
				DataPageInfo dpinfo = new DataPageInfo(atuple,tx,table);
				// int dpinfoLen = arecord.length;

				freePage(dpinfo.pageId);

			}
			// ASSERTIONS:
			// - we have freePage()'d all data pages referenced by
			// the current directory page.

			nextDirPageId = currentDirPage.getNextPage(tx,table);
			freePage(currentDirPageId);

			currentDirPageId.pageId = nextDirPageId.pageId;
			if (nextDirPageId.pageId != INVALID) {
				pinPage(currentDirPageId, currentDirPage, false);
				// currentDirPage.openHFpage(pageinbuffer);
			}
		}
		delete_db(fileName,tx);
	}

	public Scan openScan(Transaction tx) throws Exception {
		Scan newscan = new Scan(this,tx);
		return newscan;
	}

	public int getRecCnt(Transaction tx,String table) throws Exception {
		int answer = 0;
		PageID currentDirPageId = new PageID(firstDirPageID.pageId);

		PageID nextDirPageId = new PageID(0);

		HFPage currentDirPage = new HFPage();
		Page pageinbuffer = new Page();

		while (currentDirPageId.pageId != INVALID) {
			pinPage(currentDirPageId, currentDirPage, false);

			RID rid = new RID();
			Tuple atuple;
			for (rid = currentDirPage.firstRecord(tx,table); rid != null; // rid==NULL
			// means no
			// more
			// record
			rid = currentDirPage.nextRecord(rid,tx,table)) {
				atuple = currentDirPage.getRecord(rid,tx,table);
				DataPageInfo dpinfo = new DataPageInfo(atuple,tx,table);

				answer += dpinfo.reccnt;
			}

			// ASSERTIONS: no more record
			// - we have read all datapage records on
			// the current directory page.

			nextDirPageId = currentDirPage.getNextPage(tx,table);
			unpinPage(currentDirPageId, false /* undirty */);
			currentDirPageId.pageId = nextDirPageId.pageId;
		}

		// ASSERTIONS:
		// - if error, exceptions
		// - if end of heapfile reached: currentDirPageId == INVALID_PAGE
		// - if not yet end of heapfile: currentDirPageId valid

		return answer;
	} // end of getRecCnt

	public Tuple getRecord(RID rid,Transaction tx,String table) throws Exception {
		boolean status;
		HFPage dirPage = new HFPage();
		PageID currentDirPageId = new PageID();
		HFPage dataPage = new HFPage();
		PageID currentDataPageId = new PageID();
		RID currentDataPageRid = new RID();

		status = findDataPage(rid, currentDirPageId, dirPage,
				currentDataPageId, dataPage, currentDataPageRid,tx,table);

		if (status != true)
			return null; // record not found

		Tuple atuple = null;
		atuple = dataPage.getRecord(rid,tx,table);

		/*
		 * getRecord has copied the contents of rid into recPtr and fixed up
		 * recLen also. We simply have to unpin dirpage and datapage which were
		 * originally pinned by _findDataPage.
		 */

		unpinPage(currentDataPageId, false /* undirty */);

		unpinPage(currentDirPageId, false /* undirty */);

		return atuple; // (true?)OK, but the caller need check if atuple==NULL

	}

}
